package com.atguigu.bigdata.spark.zzgcore.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Classname Spark01_RDD_Operation_Transfrom
 * @Description 相同的首字母放在一个组中
 * @Date 2023/9/20 15:23
 * @Author zhuzhenguo
 */
object Spark13_RDD_Operation_Transform {
  def main(args: Array[String]): Unit = {
    // 准备环境,这个 *表示系统当前最大可用核数
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 双value类型的操作
    // 交集、并集和差集要求两个数据源类型保持一致
    // 拉链操作两个数据源的类型可以不一致
    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    val rdd2: RDD[Int] = sc.makeRDD(List(3, 4, 5, 6))
    val rdd7: RDD[String] = sc.makeRDD(List("3", "4", "5", "6"))

    // 交集
    val rdd3: RDD[Int] = rdd1.intersection(rdd2)
    println(rdd3.collect().mkString(","))
//    val rdd8 = rdd1.intersection(rdd7)
    // 并集
    val rdd4: RDD[Int] = rdd1.union(rdd2)
    println(rdd4.collect().mkString(","))
    // 差集
    val rdd5: RDD[Int] = rdd1.subtract(rdd2)
    println(rdd5.collect().mkString(","))
    // 拉链
    val rdd6: RDD[(Int, Int)] = rdd1.zip(rdd2)
    println(rdd6.collect().mkString(","))

    val rdd8: RDD[(Int, String)] = rdd1.zip(rdd7)
    println(rdd8.collect().mkString(","))
    // 关闭环境
    sc.stop()
  }
}
